package com.peng.sparktest.sparkcore

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * 扩展
 */
object SparkApiTest05_Extend {
  def main(args: Array[String]): Unit = {

    val conf: SparkConf = new SparkConf().setAppName("pv_uv_job").setMaster("local")
    val context = new SparkContext(conf)
    context.setLogLevel("ERROR")
    //数据抽样：
    //    val data: RDD[Int] = context.parallelize(1 to 100)
    //    //params： 是否抽取重复   /抽取比例（不一定绝对）/ 参考种子（种子数相同的抽取的数据一样）
    //    data.sample(withReplacement = true,0.1,seed = 200).foreach(println)
    //    println("===========================================")
    //    data.sample(withReplacement = true,0.1,seed = 200).foreach(println)
    //    println("===========================================")
    //    data.sample(withReplacement = false,0.1,seed = 200).foreach(println)
    //    println("===========================================")


    //分区数调整：
    //例如原来加载数据时设置了100个分区，但随着若干转换操作的变换（例如filter，flatmap）的操作之后，导致数据膨胀或缩小，
    //进而导致可能数据倾斜，或者每个分区实际没有几条数据，这个时候需要我们重新调整分区数量
    val data: RDD[Int] = context.parallelize(1 to 10, 5)
    println(s"data partition size:${data.getNumPartitions}")
    val data1: RDD[(Int, Int)] = data.mapPartitionsWithIndex((pIndex, pIter) => {
      pIter.map((pIndex, _))
    })
    data1.foreach(println)
    //    val data2= data1.repartition(8)
    //实际上 repartition调用的是coalesce  因此我们也可以直接调用它
    //通过coalesce可以设置是否进行shuffle操作
    //如果是分区数相对之前是增多的，那么如果shuffle为false，那么分区修改是不工作的，因为不shuffle也就无法使用分区器分区，无法确定要去的地方
    //如果是分区数相对之前是减少的，那么如果shuffle为false，那么则会走IO复制，因为这个过程是定向的
    //这里暂且大致的认为IO复制只需要把某一个分区的数据拉取到另外一个分区，然后由新分区的某一个分区指向这两个分区的数据
    //action输出时，分别调用两者的迭代器进行输出
    //    val data2: RDD[(Int, Int)] = data1.coalesce(8,shuffle = true)
    val data2: RDD[(Int, Int)] = data1.coalesce(8, shuffle = false)
    println("========================================================")
    println(s"data partition size:${data2.getNumPartitions}")
    data2.mapPartitionsWithIndex((pIndex, pIter) => {
      pIter.map((pIndex, _))
    }).foreach(println)
    while (true) {

    }
  }
}
